package com.huatai.bi.kafaka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.huatai.bi.entity.*;
import com.huatai.bi.kafaka.DTO.MqttFieldDto;
import com.huatai.bi.service.DataSetColumnService;
import com.huatai.bi.service.DirectDataSetService;
import com.huatai.bi.service.DirectoryService;
import com.huatai.bi.service.MqMqttClientService;
import com.huatai.bi.service.impl.ClickHouseJDBCService;
import com.huatai.bi.websocket.MessageResult;
import com.huatai.bi.websocket.WebSocketService;
import com.huatai.common.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @program: itc-airport-cloud
 * @description: mqtt数据处理
 * @author: songxiude
 * @create: 2024-07-17 11:20
 **/
@Service
@Slf4j
public class MqttDataAcceptService {

	@Autowired
	private MqMqttClientService mqttClientService;

	@Autowired
	private DirectoryService directoryService;

	@Autowired
	private DirectDataSetService directDataSetService;

	@Autowired
	private DataSetColumnService dataSetColumnService;

	@Autowired
	private ClickHouseJDBCService jdbcService;

	@Autowired
	private WebSocketService socketService;

	@Autowired
	private RedissonClient redissonClient;

	/**
	 * 处理mqtt数据入库
	 *
	 * @param str
	 */
	@Async("asyncServiceExecutor")
	public void dataAccept(String str) {
		JSONObject jsonObject = JSONObject.parseObject(str, Feature.OrderedField);
		String topic = jsonObject.getString("topic");
		String clientId = jsonObject.getString("clientId");
		MqMqttClientEntity mqMqttClient = mqttClientService.getOne(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, clientId));
		if(mqMqttClient == null){
			log.info("mqtt连接信息异常");
			return;
		}
		JSONObject data = jsonObject.getJSONObject("data");
		//1、判断表是否已经存在，如果不存在先创建数据集(按百度mqtt接入规范)
		String[] split = topic.split("/");
		String name = mqMqttClient.getMqttName() + "_" + mqMqttClient.getClientTopic();
		DirectoryEntity oldDir = null;
		String tablename = "";
		String markInfo = "";
		RLock lock = redissonClient.getLock(name);
		try {
			// 尝试获取锁，最多等待10秒，锁的持有时间为5秒
			boolean isLocked = lock.tryLock(10, 5, java.util.concurrent.TimeUnit.SECONDS);
			if (isLocked) {
				// 成功获取锁，执行业务逻辑
				System.out.println("获取到锁，开始执行业务");

				//根据第二段第三段作为数据表名
				tablename = "mqtt_" + split[1] + "_" + split[2] + "_" + System.currentTimeMillis();

				oldDir = directoryService.getOne(
					Wrappers.<DirectoryEntity>lambdaQuery()
						.eq(DirectoryEntity::getClientId, clientId)
						.isNotNull(DirectoryEntity::getTableName)
						.eq(DirectoryEntity::getIsDeleted, 0)
						.last("limit 1")
				);

				if(mqMqttClient.getIsParent().intValue() == 1 && StringUtil.isNotBlank(mqMqttClient.getMarkInfo())){
					//父子终端需要解析终端标识
					List<String >detailList = Arrays.asList(mqMqttClient.getMarkInfo().split("\\."));
					JSONObject detailJson = data;

					for (int i = 0; i < detailList.size(); i++) {
						Object o = detailJson.get(detailList.get(i));
						if(!(o instanceof JSONObject)){
							markInfo = o.toString();
							break;
						}
						detailJson =  (JSONObject) o;
					}
					if(StringUtil.isNotBlank(markInfo)){
						tablename = "mqtt_" + split[1] + "_" + split[2] + "_" + markInfo + "_" + System.currentTimeMillis();
						name = mqMqttClient.getMqttName() + "_" + mqMqttClient.getClientTopic() + "_" + markInfo;
						oldDir = directoryService.getOne(
							Wrappers.<DirectoryEntity>lambdaQuery()
								.eq(DirectoryEntity::getClientId, clientId)
								.eq(DirectoryEntity::getMarkInfo, markInfo)
								.isNotNull(DirectoryEntity::getTableName)
								.eq(DirectoryEntity::getIsDeleted, 0)
								.last("limit 1")
						);
					}
				}
			}
		} catch (InterruptedException e) {
			System.out.println("获取锁异常："+e.getMessage());
			e.printStackTrace();
		} finally {
			// 释放锁
			if (lock.isHeldByCurrentThread()) {
				lock.unlock();
				System.out.println("释放锁");
			}
		}

		log.info("设备名称加设备编号：" + tablename);
		log.info("是否存在数据集：" + oldDir);


		//2、满足条件创建数据集
		if (oldDir == null) {
			List<MqttFieldDto> fields = this.getFields(data,mqMqttClient);
			//1、回填tb_direct_data_set
			DirectDataSetEntity directDataSet = new DirectDataSetEntity();
			//生成id回填
			long ddsid = IdWorker.getId();
			directDataSet.setId(ddsid);
			directDataSet.setDtId(mqMqttClient.getDtId());
			directDataSet.setTableName(tablename);
			directDataSet.setType("MQTT");
			directDataSetService.save(directDataSet);
			//2、回填tb_directory
			DirectoryEntity directoryEntity = new DirectoryEntity();
			//目录id
			directoryEntity.setDtId(ddsid);
			directoryEntity.setPid(mqMqttClient.getDtId());
			directoryEntity.setName(name);
			directoryEntity.setTableName(tablename);
			directoryEntity.setIsDir(1);
			directoryEntity.setTopicInfo(topic);
			directoryEntity.setDataType(1);
			directoryEntity.setMarkInfo(markInfo);
			directoryEntity.setClientId(clientId);
			directoryService.save(directoryEntity);
			//3、回填tb_data_set_column
			boolean datasetB = this.saveDataSetColumn(ddsid, fields);
			System.out.println("创建字段结果：" + datasetB);
			//4、先创建目标表
			jdbcService.createTableSourceDS(tablename, ddsid, "");
			//更新补齐数据集id
			mqMqttClient.setDataSetId(ddsid);
			mqttClientService.updateById(mqMqttClient);
			log.info("已经生成目标表：" + tablename);

		}else{
			tablename = oldDir.getTableName();
		}
		//3、插入数据
		List<Map<String, Object>> maps = this.createData(data,mqMqttClient);
//		//ws广播数据
//		String mqdata = JSONObject.toJSONString(maps);
		String mqdata = JSONObject.toJSONString(data);
//		JSONObject object = new JSONObject();
//		object.put("mqtopic",tablename);
//		object.put("data",mqdata);
		MessageResult messageResult = new MessageResult();
		messageResult.setMsgInfo(mqdata);
		messageResult.setTopic(tablename);
		messageResult.setMsgType(3);
		socketService.sendAlls(messageResult);
		//插入数据
		jdbcService.insertDataByTableHashMap(tablename, maps);


	}

	/**
	 * 创建datasetcolumn表数据
	 *
	 * @param dsetId
	 * @param fields
	 * @return
	 */
	public boolean saveDataSetColumn(Long dsetId, List<MqttFieldDto> fields) {
		List<DataSetColumnEntity> columns = new ArrayList<>();
		for (int i = 0; i < fields.size(); i++) {
			MqttFieldDto mqttFieldDto = fields.get(i);
			DataSetColumnEntity dataSetColumnEntity = new DataSetColumnEntity();
			dataSetColumnEntity.setDataSetId(dsetId);
			dataSetColumnEntity.setColumnName(mqttFieldDto.getField());
			dataSetColumnEntity.setColumnComment(mqttFieldDto.getField());
			dataSetColumnEntity.setColumnPosition(i + 1);
			dataSetColumnEntity.setColumnType(mqttFieldDto.getFieldType());
			columns.add(dataSetColumnEntity);
		}
		return dataSetColumnService.saveBatch(columns);
	}

	/**
	 * 获取字段列表
	 *
	 * @param data
	 * @return
	 */
	public List<MqttFieldDto> getFields(JSONObject data,MqMqttClientEntity mqMqttClient) {
		List<MqttFieldDto> mqttFieldDtos = new ArrayList<>();
		if(StringUtil.isNotBlank(mqMqttClient.getDetailColumnsInfo()) || StringUtil.isNotBlank(mqMqttClient.getTopColumns())){
			List<String> topList = new LinkedList<>();
			List<String> detailList = new LinkedList<>();
			if(StringUtil.isNotBlank(mqMqttClient.getTopColumns())){
				topList = Arrays.asList(mqMqttClient.getTopColumns().split("\\."));
			}
			if(StringUtil.isNotBlank(mqMqttClient.getDetailColumnsInfo())){
				detailList = Arrays.asList(mqMqttClient.getDetailColumnsInfo().split("\\."));
			}

			//组装顶层数据字段
			for (String key : topList) {
				MqttFieldDto mqttFieldDto = new MqttFieldDto();
				Object o = data.get(key);
				if(o == null){
					mqttFieldDto.setField(key);
					mqttFieldDto.setFieldType("string");
					mqttFieldDtos.add(mqttFieldDto);
					continue;
				}

				if ("timestamp".equals(key)) {
					mqttFieldDto.setField(key + "_time");
					mqttFieldDto.setFieldType("number");

					//存储时间格式化的字段
					MqttFieldDto timeDto = new MqttFieldDto();
					timeDto.setField(key + "_date");
					timeDto.setFieldType("string");
					mqttFieldDtos.add(timeDto);
				}else{
					if (o instanceof JSONObject || o instanceof JSONArray) {
						mqttFieldDto.setField(key);
						mqttFieldDto.setFieldType("json");
					} else if (o instanceof Number) {
						mqttFieldDto.setField(key);
						mqttFieldDto.setFieldType("number");
					} else if (o instanceof Boolean) {
						mqttFieldDto.setField(key);
						mqttFieldDto.setFieldType("boolean");
					} else{
						mqttFieldDto.setField(key);
						mqttFieldDto.setFieldType("string");
					}
				}
				mqttFieldDtos.add(mqttFieldDto);
			}

			//组装明细字段数据
			JSONObject detailJson = data;
			Boolean checkStatus = false;
			for (int i = 0; i < detailList.size(); i++) {
				Object o = detailJson.get(detailList.get(i));
				if(!(o instanceof JSONObject)){
					break;
				}
				detailJson =  (JSONObject) o;
				if(i == detailList.size()-1){
					checkStatus = true;
				}
			}
			if(checkStatus){
				for (String key : detailJson.keySet()) {
					MqttFieldDto mqttFieldDto = new MqttFieldDto();
					Object o = detailJson.get(key);
					if ("timestamp".equals(key)) {
						mqttFieldDto.setField(key + "_time");
						mqttFieldDto.setFieldType("number");
					}else{
						if (o instanceof JSONObject || o instanceof JSONArray) {
							mqttFieldDto.setField(key);
							mqttFieldDto.setFieldType("json");
						} else if (o instanceof Number) {
							mqttFieldDto.setField(key);
							mqttFieldDto.setFieldType("number");
						} else if (o instanceof Boolean) {
							mqttFieldDto.setField(key);
							mqttFieldDto.setFieldType("boolean");
						} else{
							mqttFieldDto.setField(key);
							mqttFieldDto.setFieldType("string");
						}
					}
					mqttFieldDtos.add(mqttFieldDto);
				}
			}
		}else{
			for (String key : data.keySet()) {
				MqttFieldDto mqttFieldDto = new MqttFieldDto();
				Object o = data.get(key);
				if ("timestamp".equals(key)) {
					mqttFieldDto.setField(key + "_time");
					mqttFieldDto.setFieldType("number");

					//存储时间格式化的字段
					MqttFieldDto timeDto = new MqttFieldDto();
					timeDto.setField(key + "_date");
					timeDto.setFieldType("string");
					mqttFieldDtos.add(timeDto);
				}else{
					if (o instanceof JSONObject || o instanceof JSONArray) {
						mqttFieldDto.setField(key);
						mqttFieldDto.setFieldType("json");
					} else if (o instanceof Number) {
						mqttFieldDto.setField(key);
						mqttFieldDto.setFieldType("number");
					} else if (o instanceof Boolean) {
						mqttFieldDto.setField(key);
						mqttFieldDto.setFieldType("boolean");
					} else{
						mqttFieldDto.setField(key);
						mqttFieldDto.setFieldType("string");
					}
				}
				mqttFieldDtos.add(mqttFieldDto);
			}
		}
		for (int i = 0; i < mqttFieldDtos.size(); i++) {
			MqttFieldDto mqttFieldDto = mqttFieldDtos.get(i);
			if(mqttFieldDto.getField()==null){
				System.out.println("删除一个空对象");
				mqttFieldDtos.remove(i);
			}
		}

		//所有数据都增加入库时间字段，用于后续数据排序
		MqttFieldDto mqttFieldDto = new MqttFieldDto();
		mqttFieldDto.setField("resource_ts");
		mqttFieldDto.setFieldType("number");
		mqttFieldDtos.add(mqttFieldDto);
		String fields = JSON.toJSONString(mqttFieldDtos);
		log.info("组装字段：" + fields);

		return mqttFieldDtos;
	}

	public static String convertTimestampToDateTime(long timestamp) {
		// 把时间戳转换为 Instant 对象
		Instant instant = Instant.ofEpochMilli(timestamp);
		// 获取系统默认时区
		ZoneId zoneId = ZoneId.systemDefault();
		// 将 Instant 转换为 ZonedDateTime
		ZonedDateTime zonedDateTime = instant.atZone(zoneId);
		// 定义日期时间格式
		DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
		// 格式化 ZonedDateTime
		return zonedDateTime.format(formatter);
	}

	/**
	 * 组装数据字段
	 *
	 * @param data
	 * @return
	 */
	public List<Map<String, Object>> createData(JSONObject data,MqMqttClientEntity mqMqttClient) {
		List<Map<String, Object>> maps = new ArrayList<>();
		JSONArray array = null;
		Map<String, Object> map = new LinkedHashMap<>();
		if(StringUtil.isNotBlank(mqMqttClient.getDetailColumnsInfo()) || StringUtil.isNotBlank(mqMqttClient.getTopColumns())){
			List<String> topList = new LinkedList<>();
			List<String> detailList = new LinkedList<>();
			if(StringUtil.isNotBlank(mqMqttClient.getTopColumns())){
				topList = Arrays.asList(mqMqttClient.getTopColumns().split("\\."));
			}
			if(StringUtil.isNotBlank(mqMqttClient.getDetailColumnsInfo())){
				detailList = Arrays.asList(mqMqttClient.getDetailColumnsInfo().split("\\."));
			}

			//组装顶层数据字段
			for (String key : topList) {
				MqttFieldDto mqttFieldDto = new MqttFieldDto();
				Object o = data.get(key);
				if(o == null){
					map.put(key, o);
					continue;
				}
				if ("timestamp".equals(key)) {
					map.put(key + "_time", o);
					map.put(key + "_date", convertTimestampToDateTime(Long.parseLong(o.toString())));
				}else{
					map.put(key, o);
				}
			}

			//组装明细字段数据
			JSONObject detailJson = data;
			Boolean checkStatus = false;
			for (int i = 0; i < detailList.size(); i++) {
				Object o = detailJson.get(detailList.get(i));
				if(!(o instanceof JSONObject)){
					break;
				}
				detailJson =  (JSONObject) o;
				if(i == detailList.size()-1){
					checkStatus = true;
				}
			}
			if(checkStatus){
				for (String key : detailJson.keySet()) {
					Object o = detailJson.get(key);
					if ("timestamp".equals(key)) {
						map.put(key + "_time", o);
					}else{
						map.put(key, o);
					}
				}
			}
		}else{
			for (String key : data.keySet()) {
				Object o = data.get(key);
				if (o instanceof String || o instanceof JSONObject || o instanceof JSONArray) {
					map.put(key, o);

				} else if (o instanceof Long || o instanceof Integer) {
					String tmkey = null;
					if (key.equals("timestamp")) {
						tmkey = key + "_time";
						map.put(key + "_date", convertTimestampToDateTime(Long.parseLong(o.toString())));
					} else {
						tmkey = key;
					}

					map.put(tmkey, o);

				}else{
					map.put(key, o);
				}

			}
		}

		map.put("resource_ts", System.currentTimeMillis());
		maps.add(map);
		String s = JSON.toJSONString(maps);
		log.info("组装的数据：" + s);
		return maps;
	}

	/**
	 * 获取json有多少个层级
	 *
	 * @param jsonObject
	 * @return
	 */
	public int countLevels(JSONObject jsonObject) {
		int level = 1;
		for (String key : jsonObject.keySet()) {
			Object value = jsonObject.get(key);
			if (value instanceof JSONObject) {
				level += countLevels((JSONObject) value);
			}
		}
		return level;
	}
}
